package org.bjtu.hadooptest.hbaseDAO;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;


/**
 *
 */
public class BaseDAO {
    static Configuration conf;
    private HBaseAdmin handle;
    private HConnection connection;

    /**
     * @Description: 初始化配置
     * @Param: [conf]
     * @return: void
     * @Author: ztb
     * @Date: 2018/6/8
     */
    public void init() throws Exception {
        this.conf = HBaseConfiguration.create();
        this.handle = new HBaseAdmin(this.conf);
        this.connection = HConnectionManager.createConnection(conf);

    }

    /**
     * @Description: create table by table name and column families
     * @Param: [tableName, family]
     * @return: boolean
     * @Author: ztb
     * @Date: 2018/6/8
     */
    public boolean createTable(String tableName, String[] families) {
        HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
        for (String family : families) {
            desc.addFamily(new HColumnDescriptor(family));
        }
        try {
            if (this.handle.isTableAvailable(tableName)) {
                throw new Exception("table exist~");
            } else {
                this.handle.createTable(desc);
                //TODO:log
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * @Description: insert into records
     * @Param: [tableName, row, families, values]
     * @return: boolean
     * @Author: ztb
     * @Date: 2018/6/8
     */
    private void addRecord(String tableName, String rowKey, String family, List<String[]> data) throws IOException {
        Put put = new Put(Bytes.toBytes(rowKey));
        HTable table = new HTable(this.conf, Bytes.toBytes(tableName));// 单次加载
        try {
            data.forEach((d) -> put.add(family.getBytes(), d[0].getBytes(), d[1].getBytes()));
        } catch (Exception e) {
            //插入失败，检查原因
            if (!table.getTableDescriptor().getFamiliesKeys().contains(family.getBytes())) {
                System.out.println("不存在列");
            }
        }
        table.put(put);
        //TODO: log
    }

    /**
     * @Description: 插入记录
     * @Param: [tableName, rowKey, columns, values]
     * @return: void
     * @Author: ztb
     * @Date: 2018/6/8
     */
    public void addRecords(String tableName, String rowKey, String[] columns, String[] values) throws Exception {
        String family = "";
        List<String> columnsList = new ArrayList<>();
        List<String> valuesList = new ArrayList<>();
        for (int i = 0; i < columns.length; i++) {
            String[] family_column = columns[i].split(":");
            columnsList.add(family_column[1]);
            valuesList.add(values[i]);
            if (family.equals("")) {
                family = family_column[0];
            }
            if (!family.equals(family_column[0])) {
                this.addRecord(tableName, rowKey, family, columnsList, valuesList);
                columnsList.removeAll(columnsList);
                valuesList.removeAll(valuesList);
            }
            if (i == columns.length - 1) {
                this.addRecord(tableName, rowKey, family, columnsList, valuesList);
            }
        }
    }

    /**
     * @Description: 浏览表tableName某一列的数据，
     * 如果某一行记录中该列数据不存在，则返回null。
     * 要求当参数column为某一列族名称时，如果底下有若干个列限定符，
     * 要列出每个列限定符代表的列的数据；
     * 当参数column为某一列具体名称（例如“Score:Math”）时，只需要列出该列的数据。
     * @Param: [tableName, column]
     * @return: java.util.List
     * @Author: ztb
     * @Date: 2018/6/8
     */
    public Map scanColumn(String tableName, String column) throws Exception {
        String[] family_column = column.split(":");
        if (family_column.length == 1) {
            return scanFamily(tableName, family_column[0]);
        } else {
            return scanColumnData(tableName, family_column[0], family_column[1]);
        }
    }

    private Map scanFamily(String tableName, String family) throws Exception {
        HTable table = new HTable(conf, tableName.getBytes());
        Map<String, List<String>> resultMap = new HashMap<>();
        Scan scan = new Scan();
        scan.addFamily(family.getBytes());
        ResultScanner scanner = table.getScanner(scan);
        Iterator<Result> resultIterator = scanner.iterator();
        while (resultIterator.hasNext()) {
            Result result = resultIterator.next();
            NavigableMap cols = result.getFamilyMap(family.getBytes());
            NavigableSet keys = cols.navigableKeySet();
            Iterator iterator = keys.iterator();
            while (iterator.hasNext()) {
                String key = (String) iterator.next();
                if (!resultMap.containsKey(key)) {
                    List<String> col = new ArrayList<String>();
                    col.add((String) cols.get(key));
                    resultMap.put(key, col);
                } else {
                    List<String> col = resultMap.get(key);
                    col.add((String) cols.get(key));
                    resultMap.put(key, col);
                }
            }
        }
        if (resultMap.size() == 0) {
            return null;
        } else {
            return resultMap;
        }
    }

    private Map scanColumnData(String tableName, String family, String column) throws Exception {

        HTable table = new HTable(conf, tableName.getBytes());
        Map<String, List<String>> resultMap = new HashMap<>();
        Scan scan = new Scan();
        scan.addColumn(family.getBytes(), column.getBytes());
        ResultScanner scanner = table.getScanner(scan);
        Iterator<Result> resultIterator = scanner.iterator();
        List<String> colTemp = new ArrayList<>();
        while (resultIterator.hasNext()) {
            Result result = resultIterator.next();
            colTemp.add(result.getColumnLatestCell(family.getBytes(), column.getBytes()).toString());
        }
        if (colTemp.size() == 0) {
            return null;
        } else {
            resultMap.put(column, colTemp);
            return resultMap;
        }
    }

    /**
     * @Description: 修改表tableName，行row（可以用学生姓名S_Name表示），列column指定的单元格的数据。
     * @Param: [tableName, row, column]
     * @return: boolean
     * @Author: ztb
     * @Date: 2018/6/8
     */
    public void modifyData(String tableName, String row, String column, String value) throws Exception {
        String[] family_column = column.split(":");
        HTable table = new HTable(conf, tableName.getBytes());
        Put put = new Put(row.getBytes());
        put.add(family_column[0].getBytes(), family_column[1].getBytes(), value.getBytes());
        try {
            table.put(put);
            //TODO: log success
        } catch (IOException e) {
            //TODO: log failed
            System.out.println(e);
        }

    }

    /**
     * @Description: 删除表tableName中row指定的行的记录。
     * @Param: [tableName, row]
     * @return: void
     * @Author: ztb
     * @Date: 2018/6/8
     */
    public void deleteRow(String tableName, String row) throws Exception {
        HTable table = new HTable(conf, tableName.getBytes());
        Delete delete = new Delete(row.getBytes());
        try {
            table.delete(delete);
            //TODO: log success
        } catch (IOException e) {
            //TODO:failed
            System.out.println(e);
        }

    }

}
